package cn.edu.laud.sany.dataimport;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.ByteBuffer;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.zip.GZIPInputStream;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.Mutation;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

import cn.edu.laud.sany.types.DataFormatErrorException;
import cn.edu.laud.sany.types.ImportProcess;
import cn.edu.laud.sany.types.RowData;
import cn.edu.laud.sany.types.TimeSerial;
import cn.edu.laud.sany.types.UnConfiguredCFException;
import cn.edu.laud.sany.utils.MutationMap;
import cn.edu.laud.sany.utils.MyLog;
import cn.edu.laud.sany.utils.ProcessManager;
import cn.edu.laud.sany.utils.TimeUtils;

public class LaUDRunner extends Thread
{
	// constants
	public static int PROCESS_LOG_BATCH_SIZE = 10;
	public static int READ_BATCH_SIZE = 1000;
	public static int MUTATION_BATCH_SIZE = 10000;
	protected static final ConsistencyLevel CL = ConsistencyLevel.ANY;
	
	// for parse data and add timeSerial
	protected SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	protected SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd");
	// equipcode+paraId to TimeSerial map
	protected Map<String, TimeSerial> currentTimeMap = new HashMap<String, TimeSerial>();
	// timeSerial
	protected TimeSerial timeSerial = null;
	
	// for import process
	protected int threadNum;
	protected String keyspace;
	protected String nowFileName = "";
	protected long nowLineNum = 0;
	protected MyLog logger;
	
	// for thrift connection
	protected TTransport tf;
	protected TProtocol proto;
	protected Cassandra.Client client;
	protected MutationMap mutationMap;
	
	public LaUDRunner(int no, String keyspace)
	{
		this.threadNum = no;
		this.keyspace = keyspace;
		File foler = new File("log");
		if (foler.exists() == false)
		{
			foler.mkdir();
		}
		logger = MyLog.getMyLog(LaUDImport.getLogFilePath(), true);
		logger.setThreadName("Thread " + no);
		mutationMap = new MutationMap();
	}
	
	private boolean openConnection()
	{
		String ip = LaUDImport.IP[threadNum % LaUDImport.IP.length];
		try
		{
			logger.info("Try get Thrift connection...");
			tf = new TFramedTransport(new TSocket(ip, LaUDImport.laudPort));
			proto = new TBinaryProtocol(tf);
			client = new Cassandra.Client(proto);
			tf.open();
			client.set_keyspace(keyspace);
			logger.info("Get Thrift connection successful.");
		}
		catch (TTransportException e)
		{
			logger.error("Get Thrift connection error:" + e.toString() + ", LaUD IP:" + ip);
			return false;
		}
		catch (InvalidRequestException e)
		{
			logger.error("Get Thrift connection error:" + e.toString() + ", LaUD IP:" + ip);
			return false;
		}
		catch (TException e)
		{
			logger.error("Get Thrift connection error:" + e.toString() + ", LaUD IP:" + ip);
			return false;
		}
		return true;
	}
	
	private void closeConnection()
	{
		try
		{
			if (tf != null)
			{
				tf.close();
			}
			logger.info("Laud connect closed successfully.");
		}
		catch (Exception e)
		{
			logger.error(e.getMessage());
			e.printStackTrace();
		}
	}
	
	private void endImport() throws IOException, UnConfiguredCFException
	{
		if (mutationMap.mutationCount > 0)
		{
			batchImport(client, mutationMap);
		}
	}
	
	private void addRowData(RowData rowData) throws IOException, UnConfiguredCFException
	{
		Column column = new Column(ByteBufferUtil.bytes(rowData.localTime));
		column.setValue(ByteBufferUtil.bytes(rowData.paravalue));
		column.setTimestamp(System.currentTimeMillis());
		Mutation mutation = new Mutation();
		ColumnOrSuperColumn colOrSuperCol = new ColumnOrSuperColumn();
		colOrSuperCol.column = column;
		mutation.column_or_supercolumn = colOrSuperCol;
		ByteBuffer keyBuffer = ByteBufferUtil.bytes(rowData.equipCode);
		String cfName = String.format(LaUDImport.cfNameFormat, rowData.paraId);
		mutationMap.addMutation(keyBuffer, cfName, mutation);
		
		if (mutationMap.mutationCount >= MUTATION_BATCH_SIZE)
		{
			batchImport(client, mutationMap);
		}
	}
	
	private void batchImport(Cassandra.Client nowClient, MutationMap map)
			throws UnConfiguredCFException
	{
		boolean flag = false;
		while (flag == false)
		{
			try
			{
				nowClient.batch_mutate(map.getMap(), CL);
				ProcessManager.addRealImportCount(map.mutationCount);
				flag = true;
				map.clearMutation();
			}
			catch (InvalidRequestException e)
			{
				if (e.getWhy() != null && e.getWhy().startsWith("unconfigured columnfamily"))
				{
					logger.error("One batch Import error!");
					logger.error("Caused by: " + e.toString());
					logger.error("Please check you cfs file!");
					throw (new UnConfiguredCFException());
				}
				else
				{
					logger.error("One batch import error, will try again after 1 sec!");
					logger.error("Caused by: " + e.toString());
				}
			}
			catch (UnavailableException e)
			{
				logger.error("One batch import error, will try again after 1 sec!");
				logger.error("Caused by: " + e.toString());
				WaitSecond(1);
			}
			catch (TimedOutException e)
			{
				logger.error("One batch Import time out, will try again after 1 sec !");
				WaitSecond(1);
			}
			catch (TTransportException e)
			{
				logger.error("One batch import error, will try to get connection after 5 sec!");
				logger.error("Caused by: " + e.toString());
				WaitSecond(5);
			}
			catch (TException e)
			{
				logger.error("One batch import error, will try again after 1 sec!");
				logger.error("Caused by: " + e.toString());
				WaitSecond(1);
			}
		}
		
	}
	
	private long getExistingCount(String fileName, BufferedReader bReader) throws IOException
	{
		// 表头
		bReader.readLine();
		long begin = 0;
		ImportProcess process = ProcessManager.getProcessCount(fileName);
		if (process.status == 1)
		{
			begin = process.count;
			for (int k = 0; k < begin; k++)
			{
				bReader.readLine();
			}
		}
		return begin;
	}
	
	private String trimData(String oriData)
	{
		oriData = oriData.trim();
		oriData = oriData.substring(1, oriData.length() - 1);
		return oriData;
	}
	
	private RowData getRowData(String lineString) throws DataFormatErrorException
	{
		String data[] = null;
		String equipCode = null;
		int paraId = 0;
		String paravalue = null;
		long localTime = 0L;
		try
		{
			data = lineString.split(",");
			if (data.length != 6)
			{
				ProcessManager.errorDataLog(nowFileName, nowLineNum, lineString);
				return null;
			}
			equipCode = trimData(data[1]);
			paraId = Integer.valueOf(trimData(data[2]));
			if (LaUDImport.ignoreSomeParaId && LaUDImport.ignoreParaIds.contains(paraId))
			{
				return null;
			}
			paravalue = trimData(data[3]);
			String dataString = trimData(data[5]);
			try
			{
				localTime = sdf.parse(dataString).getTime();
			}
			catch (ParseException e)
			{
				try
				{
					localTime = sdf2.parse(dataString).getTime();
				}
				catch (ParseException e1)
				{
					ProcessManager.errorDataLog(nowFileName, nowLineNum, lineString);
					throw new DataFormatErrorException();
				}
			}
		}
		catch (NumberFormatException e1)
		{
			ProcessManager.errorDataLog(nowFileName, nowLineNum, lineString);
			throw new DataFormatErrorException();
		}
		// 添加流水号
		timeSerial = currentTimeMap.get(equipCode + paraId);
		if (timeSerial == null)
		{
			timeSerial = new TimeSerial(localTime, 0);
			currentTimeMap.put(equipCode + paraId, timeSerial);
		}
		else
		{
			if (localTime != timeSerial.time)
			{
				timeSerial.time = localTime;
				timeSerial.serial = 0;
			}
			else
			{
				timeSerial.serial++;
				localTime += timeSerial.serial;
				if (timeSerial.serial >= 1000)
				{
					ProcessManager.dataTooManyLog(nowFileName, nowLineNum, lineString);
				}
			}
		}
		RowData rowData = new RowData();
		rowData.paraId = paraId;
		rowData.paravalue = paravalue;
		rowData.equipCode = equipCode;
		rowData.localTime = localTime;
		return rowData;
	}
	
	private boolean importFileData(File file)
	{
		nowLineNum = 0;
		String fileName = file.getAbsolutePath();
		nowFileName = fileName;
		GZIPInputStream gzin = null;
		BufferedReader bReader = null;
		boolean stopFlag = false;
		try
		{
			try
			{
				if (LaUDImport.csvCompressed)
				{
					gzin = new GZIPInputStream(new FileInputStream(file));
					InputStreamReader isreader = new InputStreamReader(gzin);
					bReader = new BufferedReader(isreader);
				}
				else
				{
					bReader = new BufferedReader(new FileReader(file));
				}
			}
			catch (FileNotFoundException e3)
			{
				logger.error("Get BufferedReader failed: " + e3.getMessage());
				return false;
			}
			try
			{
				// 跳过已经导入的
				nowLineNum = getExistingCount(fileName, bReader);
			}
			catch (IOException e2)
			{
				logger.error("Get ExistingCount failed: " + e2.getMessage());
				bReader.close();
				return false;
			}
			long time = System.nanoTime();
			try
			{
				List<String> lineDataList = new ArrayList<String>();
				int listDataNum = 0;
				int batchNum = 0;
				boolean dataFlag = true;
				while (dataFlag && !stopFlag)
				{
					if (LaUDImport.halted)
					{
						return false;
					}
					String lineData = bReader.readLine();
					if (lineData != null)
					{
						listDataNum++;
						lineDataList.add(lineData);
						if (listDataNum < READ_BATCH_SIZE)
						{
							continue;
						}
					}
					else
					{
						dataFlag = false;
					}
					listDataNum = 0;
					List<RowData> rowDataList = new ArrayList<RowData>();
					for (String line : lineDataList)
					{
						RowData rowData = null;
						try
						{
							rowData = getRowData(line);
						}
						catch (DataFormatErrorException e)
						{
							if (LaUDImport.errorDataStop)
							{
								stopFlag = true;
								break;
							}
						}
						if (rowData != null)
						{
							rowDataList.add(rowData);
						}
						nowLineNum++;
					}
					lineDataList.clear();
					
					for (RowData rowData : rowDataList)
					{
						addRowData(rowData);
					}
					batchNum++;
					if (batchNum == PROCESS_LOG_BATCH_SIZE)
					{
						ProcessManager.updateFileCount(fileName, 1, nowLineNum);
						batchNum = 0;
					}
				}
			}
			catch (IOException e1)
			{
				stopFlag = true;
				logger.error("Error when imort file " + fileName + ", message : " + e1.getMessage());
			}
			catch (UnConfiguredCFException e2)
			{
				stopFlag = true;
				LaUDImport.halted = true;
			}
			finally
			{
				try
				{
					if (LaUDImport.csvCompressed)
					{
						gzin.close();
					}
					else
					{
						bReader.close();
					}
					if (stopFlag == false)
					{
						endImport();
					}
				}
				catch (IOException e)
				{
					stopFlag = true;
					e.printStackTrace();
				}
				catch (NullPointerException e1)
				{
					stopFlag = true;
					e1.printStackTrace();
					logger.error(e1.getMessage());
				}
			}
			logger.info(String.format("LaUDRunner end successful. File:%s; Size:%s; Time used:%s",
					file.getName(), String.valueOf(nowLineNum),
					TimeUtils.nanoSecondToString(System.nanoTime() - time)));
		}
		catch (Throwable t)
		{
			stopFlag = true;
			t.printStackTrace();
			logger.error(t.getMessage());
		}
		
		return (stopFlag == false);
	}
	
	private void WaitSecond(int secound)
	{
		try
		{
			Thread.sleep(secound * 1000);
		}
		catch (InterruptedException e)
		{
		}
	}
	
	@Override
	public void run()
	{
		File file = null;
		String fileName = null;
		Boolean waitAtFirst = true;
		Boolean halt = false;
		logger.info("Thread Start...");
		try
		{
			while (openConnection() == false)
			{
				logger.info("Begin import failed, will wait 3 sec to restart...");
				WaitSecond(3);
			}
			
			while (LaUDImport.halted == false)
			{
				// get source
				file = null;
				fileName = null;
				if (waitAtFirst)
				{
					logger.info("Wait to get available csv/gz source file...");
					waitAtFirst = false;
				}
				file = ProcessManager.getAvailableSource();
				if (file == null)
				{
					WaitSecond(10);
					continue;
				}
				waitAtFirst = true;
				fileName = file.getAbsolutePath();
				logger.info("Find a new input file: " + fileName);
				// set start import time
				if (ProcessManager.importStartTime == 0)
				{
					ProcessManager.importStartTime = System.nanoTime();
				}
				// import
				boolean succeed = importFileData(file);
				// commit task
				if (succeed)
				{
					ProcessManager.commitTask(fileName, true);
					logger.info(String.format("Import file %s succeed!", fileName));
				}
				else
				{
					ProcessManager.commitTask(fileName, false);
					logger.info(String.format("Import file %s failed!", fileName));
					if (LaUDImport.errorDataStop)
					{
						halt = true;
						break;
					}
				}
			}
		}
		catch (Exception e)
		{
			if (fileName != null)
			{
				ProcessManager.commitTask(fileName, false);
			}
			logger.error(e.getMessage());
			e.printStackTrace();
		}
		finally
		{
			closeConnection();
		}
		if (halt)
		{
			LaUDImport.stopProgram();
		}
	}
}
